package com.sirius.kafka.demo.module.kafka;


import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

@Component

public class KafkaProducer {
    private final KafkaTemplate<String, Object> kafkaTemplate;

    public KafkaProducer(KafkaTemplate<String, Object> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }


    /**
     * 设备上线
     * @param deviceCode 设备的code、在ali云那里叫deviceName
     * @author <a href="https://iiiii.run">xiaohan.yuan</a>
     */
    public void online(String deviceCode) {
        kafkaTemplate.send("deviceOnline", deviceCode);
    }

    /**
     * 设备下线
     * @param deviceCode 设备的code、在ali云那里叫deviceName
     * @author <a href="https://iiiii.run">xiaohan.yuan</a>
     */
    public void offline(String deviceCode) {
        kafkaTemplate.send("deviceOffline", deviceCode);
    }

//    @Autowired
//    private DeviceLineHandler deviceLineHandler;


//    @Value("${kichina.setting.appCloud.url}")
//    private String appCloudUrl;
//    @Value("${iot.version}")
//    private String version;


    @KafkaListener(id = "ZING",
            groupId = "${system.kafkaGroupId}",
            topics = {"66666"})
    public void onMessage(ConsumerRecord<?, ?> record) {
        String messageStr = String.valueOf(record.value());
//        IotMessage iotMessage = JSON.parseObject(messageStr, IotMessage.class);
//        if (!"CIRCULATION".equals(iotMessage.getMessageType())){
//            return;
//        }
//        List<IotMessage.PropertyListBean> propertyList = iotMessage.getPropertyList();
//
//        Optional<IotMessage.PropertyListBean> isOnline = propertyList.stream().filter(property -> StrUtil.equals(property.getName(), "online_status")).findAny();
//        if (isOnline.isPresent()){
//            MessageModel messageModel = new MessageModel();
//            messageModel.setType(100);
//            String body;
//            if ((Integer) isOnline.get().getValue() ==0){
//                messageModel.setData("OFFLINE");
//                log.info("设备【{}】下线",iotMessage.getDeviceTarget());
//                body = HttpRequest.put(appCloudUrl + "/cloud/device/status/offline/" + iotMessage.getDeviceTarget()+"/false").execute().body();
//                log.info(body);
//            }else {
//                messageModel.setData("ONLINE");
//                log.info("设备【{}】上线",iotMessage.getDeviceTarget());
//                body = HttpRequest.put(appCloudUrl + "/cloud/device/status/offline/" + iotMessage.getDeviceTarget()+"/true").execute().body();
//            }
//            log.info(body);
//            deviceLineHandler.deviceLine(iotMessage.getDeviceTarget(),JSON.toJSONString(messageModel));
//            return;
//        }
        // 是设备值
//        sendProperty(iotMessage);
//       System.out.println("来自主题{}的消息:{}",record.topic(), record.value());
    }


//    @Autowired
//    private SubscribeCache subscribeCache;
//    @Autowired
//    private SocketCache socketCache;

//    private  void sendProperty( IotMessage iotMessage){
//        String deviceName = iotMessage.getDeviceTarget();
//        CacheNode<String, Set<String>> stringListCacheNode = subscribeCache.get(deviceName);
//        if (!stringListCacheNode.isSuccess()){
//            return;
//        }
//
//        HashMap<String,TransformationMessage> dataMap = new HashMap<String,TransformationMessage>();
//        iotMessage.getPropertyList().forEach(property->{
//            if (NAME_KEYS.contains(property.getName())){;
//                property.setValue(RplStr(UnicodeUtil.toString(String.valueOf(property.getValue())),"\\",""));
//            }
//            dataMap.put(property.getName(),new TransformationMessage(property.getValue(), property.getTime()));
//        });
//        Set<String> value = stringListCacheNode.getValue();
//        log.info("更新设备{}的值:{}",iotMessage.getDeviceTarget(),JSON.toJSONString(dataMap));
//
//        MessageModel messageModel = new MessageModel();
//        messageModel.setType(200);
//        messageModel.setData(dataMap);
//        String updateData = JSON.toJSONString(messageModel);
//        for (String socketId : value) {
//            CacheNode<String, NetSocket> socketCacheNode = socketCache.get(socketId);
//            if (socketCacheNode.isSuccess()){
//                socketCacheNode.getValue().write(updateData+"&");
//            }
//        }
//    }

    //把字符str1中的str2替换为str3
    public static String RplStr(String str1,String str2,String str3){
        String strtmp="";
        int i=0,f;
        for(i=0;;i+=str2.length()){
            f=str1.indexOf(str2,i);
            if (f==-1) {
                strtmp+=str1.substring(i);
                break;
            }else{
                strtmp+=str1.substring(i,f);
                strtmp+=str3;
                i=f;
            }
        }
        return strtmp;
    }
}